package com.lagou.spark

/*
1、找到ip所属区域
http.log：用户访问网站所产生的日志。日志格式为：时间戳、IP地址、访问网址、访问数据、浏览器信息等
ip.dat：ip段数据，记录着一些ip段范围对应的位置
文件位置：data/http.log、data/ip.dat
要求：将 http.log 文件中的 ip 转换为地址。如将 122.228.96.111 转为 温州，并统计各城市的总访问量
 */
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object IPSearch {

  def ipToLong(ip: String): Long = {
    /*
    将ip转为int数字
    切分ip数据后，转为4段8位二进制数，左移位，做或运算
    第一段左移3个8位 第二段左移2个8位 第三段左移1个8位 第四段不变
     */
    val a = ip.split("\\.").map(_.toLong)
    a(0) << 24 | a(1) << 16 | a(2) << 8 | a(3)
  }

  //二分法查找指定的ip范围
  def binarySearch(arr: Array[(String, String, String)], ip: Long):Int = {
    var low =0 //最小索引
    var high = arr.length - 1 //最大索引
    while (low <= high){
      val middle = (low + high) / 2 //中间索引值
      if((ip >= arr(middle)._1.toLong) && (ip <= arr(middle)._2.toLong))
        return middle
      if(ip < arr(middle)._1.toLong){
        high = middle - 1
      }else{
        low = middle + 1
      }
    }
    -1
  }

  def main(args: Array[String]): Unit = {
    //创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    //读取ip.dat获取地址信息
    val rddIP: RDD[String] = sc.textFile("data/ip.dat")
    //对文件进行切分，获取所需字段
    val value: RDD[(String, String, String)] = rddIP.map(_.split("\\|"))
      .map(x => {
        val startIP = x(2) //起始ip
        val endIP = x(3) //结束ip
        val location = x(6) + x(7) //所属区域
        (startIP,endIP,location)
      })
    //获取切分后的IP数据
    val arrayIP: Array[(String, String, String)] = value.collect()
    //arrayIP.foreach(println)
    //将切分后的ip数据作为广播变量
    val ipBroadcast: Broadcast[Array[(String, String, String)]] = sc.broadcast(arrayIP)

    //读取http.log
    val rddHttp: RDD[String] = sc.textFile("data/http.log")
    //获取http.log中的ip字段
    val value1: RDD[(String, Int)] = rddHttp.map(_.split("\\|"))
      .map(x => {
        val uIP = x(1) //http.log中的ip
        val ipLong = ipToLong(uIP) //转为int类型的ip数
        val arrIP = ipBroadcast.value //得到广播变量中的ip数据       
		//对应查找ip所属区域
        val index = binarySearch(arrIP,ipLong)
        val local = arrIP(index)._3
        (local,1)
      })

    //聚合value1，得出结果
    val result: RDD[(String, Int)] = value1.reduceByKey(_ + _)
    result.collect().foreach(println)

    //关闭SparkContext
    sc.stop()
  }

}

